package com.qyer.log.job.sort;

 
import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.hadoop.mapreduce.LzoTextInputFormat;
import com.qyer.util.log.common.Constants;
import com.qyer.util.log.entity.AppTrackLog;
import com.qyer.util.log.entity.City;
import com.qyer.util.log.entity.Country;
import com.qyer.util.log.entity.Guide;
import com.qyer.util.log.entity.TrackLog;
import com.qyer.util.log.service.LogService;
import com.qyer.util.log.service.impl.LogService4NginxImpl;
import com.qyer.util.log.util.AppUgcUtil;
import com.qyer.util.log.util.HiveFormatUtil;
import com.qyer.util.log.util.LogFatory;
/**
 * 此类的描述： 
 * 此类的用法:
 * @author sunhongxi 
 * @version: 2015年9月15日 上午10:00:24 
 * 
 */
public class TrackSortClean
{
    //自己定义的key类应该实现WritableComparable接口
    public static class IntPair implements WritableComparable<IntPair>
    {
        String first;
        String second;
        /**
         * Set the left and right values.
         */
        public void set(String left, String right)
        {
            first = left;
            second = right;
        }
        public String getFirst()
        {
            return first;
        }
        public String getSecond()
        {
            return second;
        }
        //反序列化，从流中的二进制转换成IntPair
        public void readFields(DataInput in) throws IOException
        {
            first = in.readUTF();
            second = in.readUTF();
        }
        //序列化，将IntPair转化成使用流传送的二进制
        public void write(DataOutput out) throws IOException
        {
            out.writeUTF(first);
            out.writeUTF(second);
        }
        //重载 compareTo 方法，进行组合键 key 的比较，该过程是默认行为。
        //分组后的二次排序会隐式调用该方法。
        public int compareTo(IntPair o)
        {
            if (!first.equals(o.first) )
            {
                return first.compareTo(o.first);
            }
            else if (!second.equals(o.second))
            {
                return second.compareTo(o.second);
            }
            else
            {
                return 0;
            }
        }
 
        //新定义类应该重写的两个方法
        //The hashCode() method is used by the HashPartitioner (the default partitioner in MapReduce)
        public int hashCode()
        {
            return first.hashCode() * 157 + second.hashCode();
        }
        public boolean equals(Object right)
        {
            if (right == null)
                return false;
            if (this == right)
                return true;
            if (right instanceof IntPair)
            {
                IntPair r = (IntPair) right;
                return r.first.equals(first) && r.second.equals(second) ;
            }
            else
            {
                return false;
            }
        }
    }
    /**
      * 分区函数类。根据first确定Partition。
      */
    public static class FirstPartitioner extends Partitioner<IntPair, Text>
    {
        public int getPartition(IntPair key, Text value,int numPartitions)
        {
            return Math.abs(key.getFirst().hashCode() * 127) % numPartitions;
        }
    }
 
    /**
     * 分组函数类。只要first相同就属于同一个组。
     */
    /*//第一种方法，实现接口RawComparator
    public static class GroupingComparator implements RawComparator<IntPair> {
        public int compare(IntPair o1, IntPair o2) {
            int l = o1.getFirst();
            int r = o2.getFirst();
            return l == r ? 0 : (l < r ? -1 : 1);
        }
        //一个字节一个字节的比，直到找到一个不相同的字节，然后比这个字节的大小作为两个字节流的大小比较结果。
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2){
             return WritableComparator.compareBytes(b1, s1, Integer.SIZE/8,
                     b2, s2, Integer.SIZE/8);
        }
    }*/
    //第二种方法，继承WritableComparator
    public static class GroupingComparator extends WritableComparator
    {
        protected GroupingComparator()
        {
            super(IntPair.class, true);
        }
        //Compare two WritableComparables.
        //  重载 compare：对组合键按第一个自然键排序分组
        public int compare(WritableComparable w1, WritableComparable w2)
        {
            IntPair ip1 = (IntPair) w1;
            IntPair ip2 = (IntPair) w2;
            String l = ip1.getFirst();
            String r = ip2.getFirst();
            return l.compareTo(r);
        }
    }
    
 
 
    // 自定义map
    public static class Map extends Mapper<LongWritable, Text, IntPair, Text>
    {
    	
        private final IntPair keyPair = new IntPair();
        String[] lineArr = null;
        
        private  LogService lli = (LogService)LogFatory.getInstance(LogService4NginxImpl.class.getName());
        
        Text val = new Text();
        Text kval = new Text();
        
        static HashMap<String, String> pagetypeMaps = new HashMap<String, String>();
		static HashMap<String, String> channelMaps = new HashMap<String, String>();
		static HashMap<String, City> cityMaps = new HashMap<String, City>();
		static HashMap<String, Country> countryMaps = new HashMap<String, Country>();
		private static List<String> listUGC = new ArrayList<String>();// 存储ugc正则
		
		static HashMap<String, String> cityNameMaps = new HashMap<String, String>();
		static HashMap<String, String> countryNameMaps = new HashMap<String, String>();
		static HashMap<String, String> cateMap= new HashMap<String, String>();
		static HashMap<String, String> jn_info_Map= new HashMap<String, String>();
		static HashMap<String, String> sw_map= new HashMap<String, String>(); //id-->搜索词
		//    
		private static HashMap <String, Guide> guideMap = new HashMap<String, Guide>();
	    private static HashMap <String, String> data_result=new HashMap<String,String>();
	    private static HashMap<String,String> mapApp = new HashMap<String,String>();// 存储app规则
	    
	    //存储不合法的 device id 列表
	    private static Set<String> invalidDeviceIdSet = new HashSet<String>();
	    private static Set<String> invalidUidSet = new HashSet<String>();
		
		
		@Override
		protected void setup(Context context) throws IOException, InterruptedException {
			BufferedReader in = null;

			try {
				// 从当前作业中获取要缓存的文件
				Path[] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration());
				
				System.out.println("paths===="+paths);
				String line = null;
				String[] strArr = null;
				String platform="";
				for (Path path : paths) {
					System.out.println("path===="+path.toString());
					
					if (path.toString().contains("ugc_rule.xml")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while ((line = in.readLine()) != null) {
							String[] array = line.split("@@");
							if (array.length > 0) {
								if (!listUGC.contains(array[0])) {
									listUGC.add(array[0]);
								}

							}
						}
						
					}else if (path.toString().contains("cate_business.txt")) {
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							
							if(strArr.length<5){
								continue ;
							}
							
							cateMap.put(strArr[1]+","+strArr[2], strArr[4]);
						}
					}else if (path.toString().contains("city_country_list")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							String[] array = line.split("#\\|~");
							if (array.length != 6) {
								System.out.println(line + array.length);
								continue;
							}
							String cityStr;
							if (array[2].length() > 0 && !array[2].equals("None")) {
								cityStr = array[2];
							} else {
								cityStr = array[1];
							}
							String cityID = array[0];
							String countryStr = array[4];
							String countryID = array[3];

							cityNameMaps.put(cityStr, cityID);
							countryNameMaps.put(countryStr, countryID);
							
			                City city = new City();
			                city.setCityid(array[0]);
			                if (array[2].length() > 0 && !array[2].equals("None")) {
			                    city.setCityname(array[2].replace(" ", "_"));
			                } else {
			                    city.setCityname( array[1].replace(" ", "_"));
			                }


			                Country country = countryMaps.get(array[3]);
			                if (country == null) {
			                    country = new Country();
			                    country.setCountryid(array[3]);
			                    country.setCountryname(array[4].replace(" ", "_")); 
			                    countryMaps.put(country.getCountryid(), country);
			                }
			                city.setCountry(country);
			                cityMaps.put(city.getCityid(), city);
							
						} 
					}else if (path.toString().contains("guide_list.csv")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							String[] array = line.split("#\\|~");
			                if (array.length != 2) {
			                    continue;
			                }
			                Guide guide = new Guide();
			                guide.setId(array[0]);
			                guide.setName( array[1].replace(" ", "_"));
			                guideMap.put(guide.getId(), guide);
						}
					}else if (path.toString().contains("app_trans.xml")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							if(line.startsWith("//")){
								continue;
							}else{
								String[] array = line.split("@@");
								if (array.length > 0) {
									String api_path=array[0].split("##")[0];
									mapApp.put(api_path, array[0]);
								}
							}
						}
					}else if (path.toString().contains("device_ETL.txt")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							invalidDeviceIdSet.add(line);
						}
					}else if (path.toString().contains("uid_ETL.txt")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							invalidUidSet.add(line);
						}
					}else if (path.toString().contains("jn_info.txt")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							if(strArr.length<2){
								continue ;
							}
							jn_info_Map.put(strArr[1].toLowerCase().replace(" ", "-"), strArr[0]);
						}
					}else if (path.toString().contains("search_word_info.txt")){
						in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()),"utf-8")));
						while (null != (line = in.readLine())) {
							strArr = line.split("\t", -1);
							if(strArr.length<2){
								continue ;
							}
							sw_map.put(strArr[0], strArr[1]);
						}
					}
				}
				in.close();
			} catch (IOException e) {
				in.close();
				e.printStackTrace();
			} finally {
				try {
					if (in != null) {
						in.close();
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		
		Pattern pattern1 = Pattern.compile("^[a-zA-Z0-9-]+$");
		Pattern pattern2 = Pattern.compile("^[a-z_]+$");
		
		//根据url后去解密后的字符串
		public String getGuideContent(String url){
			String result="-";
			try {
	        	Process proc = Runtime.getRuntime().exec("python /usr/local/cloud/src/plugins/data/pycrypto.py"+" '"+url+"'");
	        	BufferedInputStream in = new BufferedInputStream(proc.getInputStream());
				BufferedReader br = new BufferedReader(new InputStreamReader(in));
				String lineStr;
				String line="";
				while ((lineStr = br.readLine()) != null) {
					line += lineStr;
				}
				br.close();
				in.close();
				result=line;
			} catch (Exception e) {
				e.printStackTrace();
			}
			return result;
		}
        
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
        {
        	try {
	            String line = value.toString();
	            InputSplit inputSplit = context.getInputSplit();
	            String fileName = ((FileSplit) inputSplit).getPath().toString();
	            
	            if(fileName!=null && fileName.contains("api_response")){
	            	AppTrackLog ll =lli.getAppTrackFormat(line);
	            	 if(ll!=null&&"/open".equals(ll.getType())){
	                	 if (ll.getDeviceid()!=null && ll.getDeviceid().length()>0 && !"".equals(ll.getDeviceid().trim())&&ll.getTrackTime()!=null&&!"".equals(ll.getTrackTime())){
	                		 //deviceid clientid 格式校验
	                         Matcher m1 = pattern1.matcher(ll.getDeviceid());
	                         Matcher m2 = pattern2.matcher(ll.getClientId());
	                		 if(m1.matches()&&m2.matches()){
	                			 
		                		 //设置UGC 相关属性 
		                		AppUgcUtil acu =new AppUgcUtil(); 
		              			java.util.Map<String,String> ugctypeMap=acu.apireplace(ll.getApiName(), ll.getInfo(), mapApp, cityMaps, countryMaps, guideMap);
		                		
		              			Long ts =Long.parseLong(ll.getTimeStamp()); 
		                		ts=ts/1000;
		                		
		                		if(invalidDeviceIdSet.contains(ll.getDeviceid())){
		                			return ;
		                		}
		                		
		                		if(invalidUidSet.contains(ll.getUserId())){
		                			return ;
		                		}
		                		
		                		StringBuilder  outStr = new StringBuilder ("");
		             			
		             			outStr.append(HiveFormatUtil.getFeild(ll.getUserId())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getDeviceid())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ts.toString())).append(Constants.FS);
		             			outStr.append("-").append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ts.toString())).append(Constants.FS);
		             			outStr.append("+08:00").append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getUserIp())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getLon())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getLat())).append(Constants.FS);
		             			
		             			if(ugctypeMap==null||ugctypeMap.get("api_path")==null){
		             				outStr.append("-").append(Constants.FS);
		             			}else{
		             				outStr.append(HiveFormatUtil.getFeild(ugctypeMap.get("api_path"))).append(Constants.FS);
		             			}
		             			
		             			outStr.append("-").append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getType())).append(Constants.FS);
		             			outStr.append("-").append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getParamsMap().get("ra_arg"))).append(Constants.FS);
//		             			outStr.append(HiveFormatUtil.getFeild(ll.getInfo())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getUserAgent())).append(Constants.FS);
		             			//outStr.append("-").append(Constants.FS);  //之前channel为-，现改为获取channle字段  王振 改
		             			
		             			/*****************王振改 start**********************/
		             			if(ugctypeMap==null||ugctypeMap.get("channel")==null){
		             				outStr.append("-").append(Constants.FS);
		             			}else{
		             				outStr.append(HiveFormatUtil.getFeild(ugctypeMap.get("channel"))).append(Constants.FS);
		             			}
		             			/*****************王振改 stop**********************/
		             			
		             			if(ugctypeMap==null||ugctypeMap.get("ugc_type")==null){
		             				outStr.append("-").append(Constants.FS);
		             			}else{
		             				outStr.append(HiveFormatUtil.getFeild(ugctypeMap.get("ugc_type"))).append(Constants.FS);
		             			}
		             			
		             			outStr.append(HiveFormatUtil.getFeild(ll.getPlatform())).append(Constants.FS);
		             			
		             			
		             			
		             			if(ugctypeMap==null||ugctypeMap.get("ugc_content")==null||"#anchor_filtrate".equals(ugctypeMap.get("ugc_content"))){
		             				outStr.append("-").append(Constants.FS);
		             			}else{
		             				outStr.append(HiveFormatUtil.getFeild(ugctypeMap.get("ugc_content"))).append(Constants.FS);
		             			}
		             			
		             			
		             			
		             			outStr.append(HiveFormatUtil.getMapFeild(ll.getExtParamMap())).append(Constants.FS);
		             			outStr.append("-").append(Constants.FS);
		             			outStr.append("-").append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getTrackTime())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getHour())).append(Constants.FS);
		             			outStr.append("-").append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getClientId())).append(Constants.FS);
		             			outStr.append(HiveFormatUtil.getFeild(ll.getInfo())).append(Constants.FS);
		             			outStr.append("-");
		             			
		                		keyPair.set(ll.getDeviceid(), ll.getTrackTime());
		                		val.set(outStr.toString()); 
		                        context.write(keyPair, val);
		                	 }else{
		                		 System.out.println(line);
		                	 }
	                	 }
	                }
	            }else{
	            	TrackLog ll =lli.getTrackFormat(line);
	            	if(ll!=null){
	            		 if (ll.getGuid()!=null && !"".equals(ll.getGuid())&&ll.getTrackTime()!=null&&!"".equals(ll.getTrackTime())){
	                 		
	                 		//设置UGC 相关属性 
	            			 ll =lli.setUGC(ll, listUGC, cityNameMaps, countryNameMaps, cateMap, jn_info_Map,sw_map);
	                 		  
	                 		StringBuilder  outStr = new StringBuilder ("");
	              			
	                 		Long ts =Long.parseLong(ll.getTimeStamp()); 
	                 		ts=ts/1000;
	                 		
	                 		if(ll.getSession_timestamp()!=null&&!"".equals(ll.getSession_timestamp())&&!"null".equals(ll.getSession_timestamp())){
	                 			Long sts =Long.parseLong(ll.getSession_timestamp()); 
	                     		sts=sts/1000;
	                     		ll.setSession_timestamp(sts.toString());
	                 		}
	                 		
	                 		String path =ll.getCurUrl(); 
	                 		if(path!=null &&path.indexOf("http://")>-1){
	                 			path=path.substring(path.indexOf("http://")+7).replaceAll("\001", "");
	                 		}
	                 		
	                 		String refer =ll.getRefUrl(); 
	                 		
	                 		if(refer!=null &&refer.indexOf("http://")>-1){
	                 			refer=refer.substring(refer.indexOf("http://")+7).replaceAll("\001", "");
	                 		}
	                 		
	                 		/*****************王振改 start 增加guideapp_page的解密**********************/
	                 		String last_content="-";
	                 		if(ll.getUgc_type().equals("guideapp_page")){
	                 			last_content=this.getGuideContent(path);
	                 			String path_tmp=path.substring(path.indexOf("page/")+5);
	                 			if(path_tmp.indexOf("/")!=-1){
	                 				path_tmp=path_tmp.substring(path_tmp.indexOf("/")+1);
	                 			}else{
	                 				path_tmp="";
	                 			}
	                 			path=path.substring(0,path.indexOf("page/")+5)+last_content+"/"+path_tmp;
	                 			
	                 		}else{
	                 			last_content=ll.getUgc_content();
	                 		}
	                 		
	                 		String last_refer_content="-";
	                 		if(ll.getRefer_ugc_type().equals("guideapp_page")){
	                 			last_refer_content=this.getGuideContent(refer);
	                 			String refer_tmp=refer.substring(refer.indexOf("page/")+5);
	                 			if(refer_tmp.indexOf("/")!=-1){
	                 				refer_tmp=refer_tmp.substring(refer_tmp.indexOf("/")+1);
	                 			}else{
	                 				refer_tmp="";
	                 			}
	                 			refer=refer.substring(0,refer.indexOf("page/")+5)+last_refer_content+"/"+refer;
	                 			
	                 		}else{
	                 			last_refer_content=ll.getRefer_ugc_content();
	                 		}
	                 		
	                 		//add by shx
	                 		if(last_content.indexOf("#anchor_filtrate")>-1){
	                 			last_content=last_content.replaceAll("#anchor_filtrate", "");
	                 		}
	                 		if(last_refer_content.indexOf("#anchor_filtrate")>-1){
	                 			last_refer_content=last_refer_content.replaceAll("#anchor_filtrate", "");
	                 		}
	                 		//
	                 		
	                 		/*****************王振改 stop**********************/
	                 		
	              			outStr.append(HiveFormatUtil.getFeild(ll.getUserId())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getGuid())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ts.toString())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getSession_timestamp())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ts.toString())).append(Constants.FS);
	              			outStr.append("+08:00").append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getUserIp())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getLon())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getLat())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(path)).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(refer)).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getType())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getCategory())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getRa_arg())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getUserAgent())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getChannel1())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getUgc_type())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getPlatform())).append(Constants.FS);
	              			/*****************王振改 start**********************/
	              			outStr.append(HiveFormatUtil.getFeild(last_content)).append(Constants.FS);
	              			/*****************王振改 stop**********************/
	              			outStr.append(HiveFormatUtil.getMapFeild(ll.getExtParamMap())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getRefer_ugc_type())).append(Constants.FS);
	              			/*****************王振改 start**********************/
	              			outStr.append(HiveFormatUtil.getFeild(last_refer_content)).append(Constants.FS);
	              			/*****************王振改 stop**********************/
	              			outStr.append(HiveFormatUtil.getFeild(ll.getTrackTime())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getHour())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getLinkUrl())).append(Constants.FS);
	              			outStr.append("-").append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getFeild(ll.getExtra())).append(Constants.FS);
	              			outStr.append(HiveFormatUtil.getMapFeild(ll.getCurlParamMap()));
	              			
	              			
	                 		keyPair.set(ll.getGuid(), ll.getTrackTime());
	                 		val.set(outStr.toString()); 
	                         context.write(keyPair, val);
	                 	 }else{
	//                 		 keyPair.set("errorline", "999999");
	//                 		 val.set("errorline 2: "+line); 
	//                          context.write(keyPair, val);
	                 	 }
	                }else{
	//                	keyPair.set("errorline", "999999");
	//            		val.set("errorline 1 : "+line); 
	//                    context.write(keyPair, val);
	                }
	            }
        	
			} catch (Exception e) {
				// TODO: handle exception
			}
        }
    }
    // 自定义reduce
    //
    public static class Reduce extends Reducer<IntPair, Text, Text, Text>
    {
        
        Text t = new Text(); 
        public void reduce(IntPair key, Iterable<Text> values,Context context) throws IOException, InterruptedException
        {
        	try {
        		List<String> res = new ArrayList<String>();
                //context.write(SEPARATOR, null);
                for (Text val : values)
                {
                	res.add(val.toString());
                }
                
                for(int i =0 ; i < res.size() ; i++){
                	if(i==(res.size()-1)){
                		t.set(res.get(i)+Constants.FS+i+"end");
                	}else{
                		t.set(res.get(i)+Constants.FS+i);
                		
                	}
                	
                	context.write(null, t);
                }
			} catch (Exception e) {
				// TODO: handle exception
			}
        }
    }
 
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException
    {
    	int returnCode=0 ;	
    	
    	System.out.println("========================== ");
    	for (String arg : args) {
    		
			System.out.println("arg : " +arg);
			
		}
    	if(args.length<6){
    		System.out.println("arg length is less than 6!" );
    		returnCode=-1;
    		return ;
    	}
    	System.out.println("========================== ");
    	
    	
    	
        // 读取hadoop配置
        Configuration conf = new Configuration();
//        conf.set("mapred.job.queue.name", "regular"); // default,regular,realtime
//        conf.set("mapred.output.compress", "true");
//        conf.set("mapred.output.compression.codec", "lzo");
        conf.set("mapreduce.reduce.memory.mb" ,"4048");
        conf.set("mapreduce.reduce.java.opts" ,"-Xmx3276m");
        
        // 实例化一道作业
        Job job = new Job(conf, "TrackSortClean");
        job.setJarByClass(TrackSortClean.class);
        // Mapper类型
        job.setMapperClass(Map.class);
        // 不再需要Combiner类型，因为Combiner的输出类型<Text, IntWritable>对Reduce的输入类型<IntPair, IntWritable>不适用
        //job.setCombinerClass(Reduce.class);
        // Reducer类型
        job.setReducerClass(Reduce.class);
        // 分区函数
        job.setPartitionerClass(FirstPartitioner.class);
        // 分组函数
        job.setGroupingComparatorClass(GroupingComparator.class);
 
        // map 输出Key的类型
        job.setMapOutputKeyClass(IntPair.class);
        // map输出Value的类型
        job.setMapOutputValueClass(Text.class);
        // rduce输出Key的类型，是Text，因为使用的OutputFormatClass是TextOutputFormat
        job.setOutputKeyClass(Text.class);
        // rduce输出Value的类型
        job.setOutputValueClass(Text.class);
 
        // 将输入的数据集分割成小数据块splites，同时提供一个RecordReder的实现。
        job.setInputFormatClass(LzoTextInputFormat.class);
        // 提供一个RecordWriter的实现，负责数据输出。
        job.setOutputFormatClass(TextOutputFormat.class);
        
        job.setNumReduceTasks(3);
        
        
        String ugctypePath= args[3];
        
        
        //设置字典文件
        // 字典
		FileSystem hdfs1 = FileSystem.get(conf);
		FileStatus[] fileList = hdfs1.listStatus(new Path(ugctypePath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("ugc_rule")) {
				System.out.println("----------------->>> " + ugctypePath + "/" + fileName);
				String inPath2Link = new Path(ugctypePath + "/" + fileName).toUri().toString() + "#"+ "ugc_rule.xml" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		String cateMapPath= args[7];
        
		// category busniess字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(cateMapPath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("cate_business.txt")) {
				System.out.println("----------------->>> " + cateMapPath + "/" + fileName);
				String inPath2Link = new Path(cateMapPath + "/" + fileName).toUri().toString() + "#"+ "cate_business.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
	
		
		String areaPath= args[4];
        
        //设置字典文件
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(areaPath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("city_country_list")) {
				System.out.println("----------------->>> " + areaPath + "/" + fileName);
				String inPath2Link = new Path(areaPath + "/" + fileName).toUri().toString() + "#"+ "city_country_list.csv" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		String guidePath= args[5];
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(guidePath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("guide_list.csv")) {
				System.out.println("----------------->>> " + guidePath + "/" + fileName);
				String inPath2Link = new Path(guidePath + "/" + fileName).toUri().toString() + "#"+ "guide_list.csv" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		String apptransPath= args[6];
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(apptransPath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("app_trans")) {
				System.out.println("----------------->>> " + apptransPath + "/" + fileName);
				String inPath2Link = new Path(apptransPath + "/" + fileName).toUri().toString() + "#"+ "app_trans.xml" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		String deviceidpath= args[8];
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(deviceidpath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("device_ETL.txt")) {
				System.out.println("----------------->>> " + deviceidpath + "/" + fileName);
				String inPath2Link = new Path(deviceidpath + "/" + fileName).toUri().toString() + "#"+ "device_ETL.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		
		String uidpath= args[9];
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(uidpath));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("uid_ETL.txt")) {
				System.out.println("----------------->>> " + uidpath + "/" + fileName);
				String inPath2Link = new Path(uidpath + "/" + fileName).toUri().toString() + "#"+ "uid_ETL.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}

		String jn_path = args[10];
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(jn_path));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("jn_info.txt")) {
				System.out.println("----------------->>> " + jn_path + "/" + fileName);
				String inPath2Link = new Path(jn_path + "/" + fileName).toUri().toString() + "#"+ "jn_info.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
		String search_word_path = args[11];
        // 字典
		hdfs1 = FileSystem.get(conf);
		fileList = hdfs1.listStatus(new Path(search_word_path));
		for (int i = 0; i < fileList.length; i++) {
			String fileName = fileList[i].getPath().getName();
			if (fileName.contains("search_word_info.txt")) {
				System.out.println("----------------->>> " + search_word_path + "/" + fileName);
				String inPath2Link = new Path(search_word_path + "/" + fileName).toUri().toString() + "#"+ "search_word_info.txt" + i;
				DistributedCache.addCacheFile(new URI(inPath2Link), job.getConfiguration());
			}
		}
 
        // 输入hdfs路径
		FileInputFormat.addInputPath(job, new Path(args[0] ));
		FileInputFormat.addInputPath(job, new Path(args[1]));
        
        // 输出hdfs路径
        FileSystem.get(conf).delete(new Path(args[2]), true);
        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        
        //启用压缩
        FileOutputFormat.setCompressOutput(job, true);
        //lzo格式
        FileOutputFormat.setOutputCompressorClass(job, com.hadoop.compression.lzo.LzopCodec.class); 
        // 提交job
        returnCode=job.waitForCompletion(true) ? 0 : 1;
        
        System.exit(returnCode);
    }
}